/*
 * MIT License
 *
 * Copyright (c) 2024 sky2080
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
package com.sky.disruptor;

import com.sky.disruptor.handler.DemoMessageHandler;
import com.sky.disruptor.handler.TypeBMessageHandler;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;

/**
 * disruptor 自定义核心处理工具
 * @param <T>
 */
@Slf4j
public class SkyDisruptorCoreProcessor<T> {
  private static final String THREAD_NAME_PREFIX = "disruptor-queue-";

  private final DisruptorQueueHandler<T> parallelQueueHandler;


  private Map<String, DisruptorEventListener<T>> registerHandler(List<Object> disruptorMqListenerClasses) {
    Map<String, DisruptorEventListener<T>> map = new HashMap<>();
    if(CollectionUtils.isEmpty(disruptorMqListenerClasses)){
      Stream.of(new DemoMessageHandler(), new TypeBMessageHandler()).forEach(handler->{
        // 使用ServiceLoader或自定义的类扫描机制来查找带有@SkyDisruptorListener的类
        // 通过反射获取注解并注册处理器
        if (handler.getClass().isAnnotationPresent(SkyDisruptorMQListener.class)) {
          SkyDisruptorMQListener annotation = handler.getClass().getAnnotation(SkyDisruptorMQListener.class);
          map.put(annotation.topic(), (DisruptorEventListener<T>) handler);
        }
      });
    } else {
      disruptorMqListenerClasses.stream().forEach(handler->{
        if (handler.getClass().isAnnotationPresent(SkyDisruptorMQListener.class)) {
          SkyDisruptorMQListener annotation = handler.getClass().getAnnotation(SkyDisruptorMQListener.class);
          map.put(annotation.topic(), (DisruptorEventListener<T>) handler);
        }
      });
    }

    return map;
  }


  public SkyDisruptorCoreProcessor(List<Object> disruptorMqListenerClasses) {
    Map<String, DisruptorEventListener<T>> disruptorEventTopicListenerMap = registerHandler(disruptorMqListenerClasses);
    DisruptorQueueHandler.Builder<T> builder = new DisruptorQueueHandler.Builder<T>()
      .setBufferSize(1024 * 16)
      .setThreads(Runtime.getRuntime().availableProcessors())
      .setProducerType(ProducerType.MULTI)
      .setNamePrefix(THREAD_NAME_PREFIX)
      .setWaitStrategy(new SleepingWaitStrategy());
    // new BusySpinWaitStrategy()
    // new YieldingWaitStrategy()
    // new SleepingWaitStrategy()
    // new BlockingWaitStrategy()
    builder.setListeners(disruptorEventTopicListenerMap);
    this.parallelQueueHandler = builder.build();
  }

  public void process(String topic, T event) {
    this.parallelQueueHandler.add(topic, event);
  }

  public void start() {
    parallelQueueHandler.start();
  }

  public void shutdown() {
    parallelQueueHandler.shutDown();
  }


}


